package com.salmon.study.actors;


import akka.actor.ActorRef;
import akka.actor.UntypedAbstractActor;
import com.salmon.study.message.MapData;
import com.salmon.study.message.WordCount;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.StringTokenizer;

@Slf4j
public class MapActor extends UntypedAbstractActor {

    private ActorRef reduceActor;

    private String[] STOP_WORDS = {"a", "is"};

    private List<String> STOP_WORDS_LIST = Arrays.asList(STOP_WORDS);

    public MapActor(ActorRef inReduceActor) {
        this.reduceActor = inReduceActor;
    }

    @Override
    public void onReceive(Object message) throws Throwable {
        log.info("{}", message);
        if(message instanceof String) {
            String work = String.valueOf(message);
            MapData data = evaluateExpression(work);
            this.reduceActor.tell(data, ActorRef.noSender());
        } else {
            unhandled(message);
        }
    }

    private MapData evaluateExpression(String line) {
        List<WordCount> dataList = new ArrayList<>();
        StringTokenizer parser = new StringTokenizer(line);
        while(parser.hasMoreTokens()){
            String word = parser.nextToken().toLowerCase();
            if(!STOP_WORDS_LIST.contains(word)) {
                dataList.add(new WordCount(word, 1));
            }
        }
        return new MapData(dataList);
    }
}
